package com.chj.design.producerconsumer;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

// https://juejin.cn/post/6844903512443224072#heading-3
public class ProducerConsumer5 {

	// 抽象关键的接口
	public interface Consumer {
		void consume() throws InterruptedException;
	}

	public interface Producer {
		void produce() throws InterruptedException;
	}

	// 抽象类
	static abstract class AbstractConsumer implements Consumer, Runnable {
		@Override
		public void run() {
			while (true) {
				try {
					consume();
				} catch (Exception e) {
					// TODO: handle exception
					e.printStackTrace();
					break;
				}
			}
		}
	}

	static abstract class AbstractProducer implements Producer, Runnable {
		@Override
		public void run() {
			while (true) {
				try {
					produce();
				} catch (Exception e) {
					// TODO: handle exception
					e.printStackTrace();
					break;
				}
			}
		}
	}

	// 生产者 消费者抽象工厂方法
	public interface Model {
		Runnable newRunnableConsumer();

		Runnable newRunnableProducer();
	}

	public static class Task {
		public int no;

		public Task(int no) {
			this.no = no;
		}
	}

	public static class BlockingQueueModel implements Model {
		private final BlockingQueue<Task> queue;
		private final AtomicInteger increTaskNo = new AtomicInteger(0);

		public BlockingQueueModel(int cap) {
			this.queue = new LinkedBlockingQueue<>(cap);
		}

		@Override
		public Runnable newRunnableConsumer() {
			return new ConsumerImpl();
		}

		@Override
		public Runnable newRunnableProducer() {
			return new ProducerImpl();
		}

		private class ConsumerImpl extends AbstractConsumer implements Consumer, Runnable {
			@Override
			public void consume() throws InterruptedException {
				Task task = queue.take();
				Thread.sleep(500 + (long) (Math.random() * 500));
				System.out.println("consume: " + task.no);
			}
		}

		private class ProducerImpl extends AbstractProducer implements Producer, Runnable {
			@Override
			public void produce() throws InterruptedException {
				Thread.sleep((long) (Math.random() * 1000));

				Task task = new Task(increTaskNo.getAndIncrement());
				System.out.println("produce: " + task.no);
				queue.put(task);
			}
		}
	}

	public static void main(String[] args) {
		Model model = new BlockingQueueModel(3);
		for (int i = 0; i < 2; i++) {
			new Thread(model.newRunnableConsumer()).start();
		}
		for (int i = 0; i < 5; i++) {
			new Thread(model.newRunnableProducer()).start();
		}
	}

}